package com.xdja.kafka.file.sink;

import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.connect.connector.Task;
import org.apache.kafka.connect.sink.SinkConnector;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 从kafka的topic中读取文件内容，并写入到文件的connector类
 */
public class FileSinkConnector extends SinkConnector {
    // 声明文件配置变量
    public static final String FILE_CONFIG = "file";
    // 实例化一个配置对象   properties
    //    file=文件名
    private static final ConfigDef CONFIG_DEF = new ConfigDef().define(FILE_CONFIG, ConfigDef.Type.STRING, ConfigDef.Importance.HIGH, "Destination filename.");

    // 声明一个文件名变量
    private String filename;

    /**
     * 构造方法
     */
    public FileSinkConnector(){
        System.out.println(  "FileSinkConnector构造方法");
    }

    /**
     * 获取版本信息
     * @return
     */
    public String version() {
        String version="1.0.0";
        System.out.println("调用了FileSinkConnector的version() "+version);
        return version;
    }

    /**
     * connector启动时记载的方法
     * @param props
     */
    public void start(Map<String, String> props) {
        filename = props.get(FILE_CONFIG);
        System.out.println("调用了FileSinkConnector的start() "+props);
    }

    /**
     *指定与本connector关联的任务处理类
     */
    public Class<? extends Task> taskClass() {
        System.out.println("调用了FileSinkConnector的taskClass() ");
        return FileSinkTask.class;
    }

    /**
     * 设置任务配置信息
     * @param maxTasks
     * @return
     */
    public List<Map<String, String>> taskConfigs(int maxTasks) {
        System.out.println("调用了FileSinkConnector的taskConfigs() "+maxTasks);
        ArrayList<Map<String, String>> configs = new ArrayList<Map<String, String>>();
        for (int i = 0; i < maxTasks; i++) {
            Map<String, String> config = new HashMap<String, String>();
            if (filename != null)
                config.put(FILE_CONFIG, filename);
            configs.add(config);
        }
        return configs;
    }

    /**
     * connecotr停止时加载的方法
     */
    public void stop() {
        System.out.println("调用FileSinkConnector的stop()");
    }

    /**
     * 获取配置内容
     * @return
     */
    public ConfigDef config() {
        return CONFIG_DEF;
    }
}
